#pragma once
#include <set>
#include <map>
#include <chrono>
#include <stdint.h>
#include <functional>
#include "packet_parser.h"
#include "socket_session.h"
#include "subscribe_manager.h"
#include "const_value.h"
#include "packet_msg_.h"
#include "config.h"


namespace acl
{
namespace mqtt
{
class client_session : public std::enable_shared_from_this<client_session>
{

public:
enum connection_status_t
{
	e_ok = 0,
	e_offline = 1,
};
client_session ()
{}

client_session &ack_connect_ok (connect_msg &&msg)
{
	client_session_ = msg.connect_flags_.clean_session_? true:false;
	keepalive_ = msg.keep_alive_;
	if (msg.connect_flags_.will_flag_) {
		will_msg.fixed_header_.control_packet_type_ = e_publish;
		will_msg.fixed_header_.DUP_flag_ = 0;
		will_msg.fixed_header_.QoS_level_ = msg.connect_flags_.will_qoS_;
		will_msg.fixed_header_.RETAIN_ = msg.connect_flags_.will_retain_;
		will_msg.packet_identifier_ = 0;
		will_msg.topic_name_ = msg.will_topic_;
		will_msg.payload_.reset (new std::string (msg.will_message_));

		will_msg.fixed_header_.remaining_length_ =
			msg.will_message_.size () + will_msg.topic_name_.size () + 2 + 2;
		has_will_msg_ = true;
	}
	connection_status_ = e_ok;
	connect_ack_msg ack;
	ack.return_code_ = 0x00;
	ack.session_present_flag_ = sessoin_reuse_ ? 0x00:0x00;

	msg_buffers_.clear ();
	is_sending_buf_ = false;

	send_msg (packet_msg (std::move (ack)));
	socket_session_.to_recv ();
	return *this;
}
client_session &bind_packet_parser (packet_parser &&_parser)
{
	packet_parser_ = std::move (_parser);
	init_packet_parser ();
	return *this;
}
client_session &bind_socket_session (socket_session && _socket_session)
{
	socket_session_ = std::move (_socket_session);
	socket_session_.bind_recv_callback ([this](char *data, int len) {
		if (packet_parser_.do_parse (data, len) == false)
			return false;
		socket_session_.to_recv ();
		return true;

	}).bind_send_callback ([this](socket_session &){
		countine_send ();
		return true;
	}).bind_disconnect_callback([this](socket_session &) { 
		session_offline();
		return false;
	});
	return *this;
}

std::string get_id ()
{
	return sessionId_;
}

client_session &bind_session_id (std::string sessionId)
{
	sessionId_ = sessionId;
	return *this;
}

void recv_msg (publish_msg msg, uint8_t retain)
{
	if (msg.fixed_header_.QoS_level_ == 0 && 
		connection_status_ == e_offline)
		return ;

	msg.fixed_header_.DUP_flag_ = 0;
	msg.fixed_header_.RETAIN_ = retain;
	publish_msgs_.emplace_back(std::move(msg));
	if ((is_sending_buf_== false) && 
		(connection_status_ == e_ok))
		countine_send ();
}


void session_reuse (bool reuse)
{
	sessoin_reuse_ = reuse;
}
void check_alive ()
{
	return;

	if (connection_status_ == e_offline)
		return;
	auto now = std::chrono::high_resolution_clock::now ();
	auto diff = std::chrono::duration_cast<std::chrono::seconds>
		(now - last_msg_timep_).count ();

	if (keepalive_ && diff > keepalive_) {
		session_offline ();
	}
}

private:
	void session_offline ()
	{
		is_sending_buf_ = false;
		socket_session_.destory ();
		connection_status_ = e_offline;
		msg_buffers_.clear ();

		auto itr = std::remove_if(publish_msgs_.begin(),
								  publish_msgs_.end(),
								  [](publish_msg &msg) {
			return msg.fixed_header_.QoS_level_ == 0;
		});
		publish_msgs_.erase(itr, publish_msgs_.end());

		std::for_each(published_msgs_.begin(),
					  published_msgs_.end(),
					  [](publish_msg& msg) {
			msg.fixed_header_.DUP_flag_ = 1;
		});
		published_msgs_.insert (published_msgs_.end (), 
								publish_msgs_.begin (), 
								publish_msgs_.end ());
		publish_msgs_.swap (published_msgs_);
		published_msgs_.clear ();
	}
	void init_packet_parser ()
	{
		packet_parser_.
			bind_subscribe_msg_handler ([this](subscribe_msg &&msg) {
			handle_subscribe_msg(msg);
			return true;
		})
		.bind_ping_req_msg_handler ([this](ping_req_msg &&msg) {
			(void) msg;
			ping_rsp_msg rsp;
			rsp.fixed_header_.control_packet_type_ = e_ping_rsp;
			rsp.fixed_header_.reserved_ = 0;
			rsp.fixed_header_.remaining_length_ = 0;
			send_msg (packet_msg (rsp));
			last_msg_timep_ = std::chrono::high_resolution_clock::now ();
			return true;

		})
		.bind_publish_msg_handler ([this](publish_msg &&msg) {
			handle_publish_msg( std::move(msg));
			return true;
		})
		.bind_publish_ack_msg_handler ([this](publish_ack_msg &&msg) {
			handle_publish_ack_msg (msg);
			return true;
		})
		.bind_publish_rec_msg_handler ([this](publish_rec_msg &&msg) {
			publish_rel_msg rel;
			rel.packet_identifier_ = msg.packet_identifier_;
			send_msg (packet_msg (rel));
			return true;
		})
		.bind_publish_rel_msg_handler ([this](publish_rel_msg &&msg) {
			publish_comp_msg comp;
			comp.packet_identifier_ = msg.packet_identifier_;
			send_msg (packet_msg (comp));
			return true;

		}).bind_publish_comp_msg_handler ([this](publish_comp_msg &&msg) {
			handle_publish_comp_msg (msg);
			return true;

		}).bind_unsubscribe_msg_handler ([this](unsubscribe_msg &&msg) {
			handle_unsubscribe_msg (msg);
			return true;
		}).bind_disconnect_msg_handler ([this](disconnect_msg &&msg) {
			(void) msg;
			session_offline ();
			return false;
		});

	}

	void handle_publish_msg(publish_msg &&msg)
	{
		if (msg.fixed_header_.RETAIN_){
			msg_storer::get_instance ().store_msg (msg);
		}
		subscribe_manager::get_instance().pub_msg(std::move(msg));
		last_msg_timep_ = std::chrono::high_resolution_clock::now();
		auto qos = msg.fixed_header_.QoS_level_;
		if (qos == 0)
			return;
		else if(qos == 1){
			publish_ack_msg ack;
			ack.packet_identifier_ = msg.packet_identifier_;
			send_msg (packet_msg (ack));
		}
		else {
			publish_rec_msg rec;
			rec.packet_identifier_ = msg.packet_identifier_;
			send_msg (packet_msg (rec));
		}
	}

	void handle_publish_ack_msg (const publish_ack_msg &msg)
	{
		if (msg.packet_identifier_==0)
			return;
		for (auto it = published_msgs_.begin ();
			it != published_msgs_.end (); ++it) {
			if (it->packet_identifier_ == msg.packet_identifier_) {
				published_msgs_.erase (it);
				return;
			}
		}
	}
	
	void handle_publish_comp_msg (const  publish_comp_msg & comp)
	{
		for (auto it = published_msgs_.begin ();
		it != published_msgs_.end (); ++it) {
			if (it->packet_identifier_ == comp.packet_identifier_) {
				published_msgs_.erase (it);
				return;
			}
		}
	}

	void handle_unsubscribe_msg (const unsubscribe_msg &msg)
	{
		unsubscribe_ack_msg ack;
		ack.packet_identifier_ = msg.packet_identifier_;

		for (auto &itr : msg.topic_filters_)
		{
			if (topic_fiters_.find (itr) != topic_fiters_.end()) {
				subscribe_manager::get_instance ().unsub (itr, clientId_);
			}
		}

		send_msg (packet_msg(ack));
	}

	void handle_subscribe_msg (const subscribe_msg &msg)
	{
		subscribe_ack_msg ack;
		ack.fixed_header_.control_packet_type_ = e_subscribe_ack;
		ack.fixed_header_.reserved_ = 0;
		ack.packet_identifier_ = msg.packet_identifier_;

		const int mcount = config::get_instance().max_subscribe_count();
		auto &subm = subscribe_manager::get_instance();
		for (auto &itr : msg.topic_filters) {

			if (mcount > 0) {
				if ((int)topic_fiters_.size () < mcount){
					topic_fiters_[itr.topic_] = itr;
					if(subm.sub (shared_from_this (), itr))
						ack.return_code_.push_back (itr.requested_QoS_);
					else
						ack.return_code_.push_back(0x08);
				}
				else {
					ack.return_code_.push_back(0x08);
				}
			}else{
				if(subm.sub (shared_from_this (), itr))
					ack.return_code_.push_back (itr.requested_QoS_);
				else
					ack.return_code_.push_back(0x08);
			}
		}
		ack.fixed_header_.remaining_length_ = ack.return_code_.size() + 2;
		send_msg (packet_msg (ack));
	}

	void send_msg (std::string &&msg)
	{
		if (connection_status_ == e_ok &&
			is_sending_buf_ == false) {

			assert (msg_buffers_.empty ());
			if (msg.size () < MAX_SEND_SIZE) {
				socket_session_.send (std::move (msg));
			}
			else {
				msg_buffers_.push_back ({MAX_SEND_SIZE, std::move (msg)});
				socket_session_.send (msg.data (), MAX_SEND_SIZE);
			}
			return;
		}
		msg_buffers_.push_back ({0, std::move (msg)});
	}

	void countine_send ()
	{
	Send:
		if(msg_buffers_.empty())
			reload_msg_buffer();

		if (msg_buffers_.empty ()) {
			is_sending_buf_ = false;
			return;
		}
		auto &msg_buf = msg_buffers_.front ();
		if(msg_buf.send_pos == (int) msg_buf.buffer.size()) {
			msg_buffers_.pop_front();
			goto Send;
		}
		int remain = msg_buf.buffer.size () - msg_buf.send_pos;
		int min_size = std::min<int> (MAX_SEND_SIZE, remain);
		int send_pos = msg_buf.send_pos;
		
		msg_buf.send_pos += min_size;
		is_sending_buf_ = true;

		socket_session_.send(msg_buf.buffer.data()+send_pos, min_size);
	}

	void reload_msg_buffer() 
	{
		if(publish_msgs_.empty())
			return;
		auto &msg = publish_msgs_.front();
		if(msg.fixed_header_.QoS_level_)
			msg.packet_identifier_ = packet_identifier_++;

		msg.fixed_header_.remaining_length_ = 
			(msg.payload_ ? msg.payload_->size() : 0) +
			(msg.fixed_header_.QoS_level_ ? 2 : 0) +
			(2 + msg.topic_name_.size());

		if(msg.fixed_header_.QoS_level_)
			published_msgs_.push_back (msg);
		msg_buffers_.emplace_back(0, packet_msg(msg));
		publish_msgs_.pop_front();
	}
private:
	struct msg_buffer 
	{
		msg_buffer (int _send_pos, const std::string & buf)
		:send_pos(_send_pos),
			buffer(buf)
		{
		}
		int send_pos;
		std::string buffer;
	};

	std::string sessionId_;
	std::string clientId_;

	socket_session socket_session_;
	packet_parser packet_parser_;
	
	publish_msg will_msg;
	bool has_will_msg_ = false;
	bool is_sending_buf_ = false;
	bool client_session_ = false;
	bool sessoin_reuse_ = false;
	uint16_t keepalive_ = 0;
	uint16_t packet_identifier_ = 1;
	std::list<publish_msg> publish_msgs_;
	std::list<publish_msg> published_msgs_;
	std::list<msg_buffer> msg_buffers_;
	connection_status_t connection_status_ = e_ok;
	std::map<std::string, subscribe_msg::topic_filter> topic_fiters_;
	std::chrono::high_resolution_clock::time_point last_msg_timep_ 
		= std::chrono::high_resolution_clock::now();
};
}
}
